package com.itheima.search.service;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
import com.itheima.search.domain.NewsDocument;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author NicelYHHello
 */
@Service
public class NewsSyncService {

    private final JdbcTemplate jdbcTemplate;
    private final ElasticsearchClient elasticsearchClient;

    public NewsSyncService(JdbcTemplate jdbcTemplate, ElasticsearchClient elasticsearchClient) {
        this.jdbcTemplate = jdbcTemplate;
        this.elasticsearchClient = elasticsearchClient;
    }

    /**
     * 全量同步
     */
    public void fullSync() {
        String sql = "SELECT * FROM tb_news WHERE status = 1";
        syncData(sql);
    }

    /**
     * 执行数据同步
     */
    private void syncData(String sql, Object... params) {
        try {
            // 从MySQL查询数据
            List<Map<String, Object>> newsList = jdbcTemplate.queryForList(sql, params);

            if (newsList.isEmpty()) {
                System.out.println("没有需要同步的数据");
                return;
            }

            // 准备批量操作
            List<BulkOperation> operations = new ArrayList<>();

            for (Map<String, Object> news : newsList) {
                NewsDocument document = convertToDocument(news);

                IndexOperation<NewsDocument> indexOperation = IndexOperation.of(io ->
                        io.index("tb_news")
                                .id(document.getNewsId().toString())
                                .document(document)
                );

                operations.add(BulkOperation.of(bo -> bo.index(indexOperation)));
            }

            // 批量提交到ES
            BulkRequest bulkRequest = BulkRequest.of(br -> br.operations(operations));
            BulkResponse response = elasticsearchClient.bulk(bulkRequest);

            if (response.errors()) {
                // 处理错误
                response.items().forEach(item -> {
                    if (item.error() != null) {
                        System.err.println("Error indexing document: " + item.error().reason());
                    }
                });
                throw new RuntimeException("ES批量操作存在错误");
            }

            System.out.println("成功同步 " + newsList.size() + " 条数据");

        } catch (IOException e) {
            System.err.println("同步数据到ES失败: " + e.getMessage());
            throw new RuntimeException("同步数据到ES失败", e);
        }
    }

    /**
     * 将数据库记录转换为ES文档
     */
    private NewsDocument convertToDocument(Map<String, Object> news) {
        NewsDocument document = new NewsDocument();

        // 使用安全的方法获取和转换值
        document.setNewsId(getLongSafe(news, "news_id"));
        document.setTitle(getStringSafe(news, "title"));
        document.setTypeId(getIntegerSafe(news, "type_id"));
        document.setContent(getStringSafe(news, "content"));
        document.setDelayTime(getByteSafe(news, "delay_time"));
        document.setSource(getStringSafe(news, "source"));
        document.setCreateBy(getStringSafe(news, "create_by"));
        document.setCreateTime(getTimestampSafe(news, "create_time"));
        document.setUpdateBy(getStringSafe(news, "update_by"));
        document.setUpdateTime(getTimestampSafe(news, "update_time"));
        document.setStatus(getShortSafe(news, "status"));
        document.setRemark(getStringSafe(news, "remark"));
        document.setNoticeId(getIntegerSafe(news, "notice_id"));

        return document;
    }

    // 添加辅助方法进行安全转换
    private Long getLongSafe(Map<String, Object> map, String key) {
        Object value = map.get(key);
        return (value instanceof Number) ? ((Number) value).longValue() : null;
    }

    private String getStringSafe(Map<String, Object> map, String key) {
        Object value = map.get(key);
        return (value != null) ? value.toString() : null;
    }

    private Integer getIntegerSafe(Map<String, Object> map, String key) {
        Object value = map.get(key);
        return (value instanceof Number) ? ((Number) value).intValue() : null;
    }

    private Short getShortSafe(Map<String, Object> map, String key) {
        Object value = map.get(key);
        return (value instanceof Number) ? ((Number) value).shortValue() : null;
    }

    private Byte getByteSafe(Map<String, Object> map, String key) {
        Object value = map.get(key);
        return (value instanceof Number) ? ((Number) value).byteValue() : null;
    }

    private LocalDateTime getTimestampSafe(Map<String, Object> map, String key) {
        Object value = map.get(key);
        if (value instanceof java.sql.Timestamp) {
            return ((java.sql.Timestamp) value).toLocalDateTime();
        }
        return null;
    }
}